package cn.ryh.kafka.code.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Objects;
import java.util.Properties;
import java.util.stream.IntStream;

/**
 * 带回调函数的异步发送
 *
 * @author Raoyh
 * @Description
 * @create 2022-07-26 9:22
 */
@Slf4j
public class CustomProducerCallback {
    public static void main(String[] args) {
        // 1、创建配置对象
        Properties properties = new Properties();
        // 2、给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "119.3.50.65:8092");
        // 3、配置 key,value 序列化（必须）
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 4、创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        // 5、调用 send 方法,发送消息
        IntStream.range(0, 10).forEach((i) -> {

            kafkaProducer.send(new ProducerRecord<>("first", "kafkaCallback" + i), (metadata, e) -> {
                // 该方法在 Producer 收到 ack 时调用，为异步调用
                if (Objects.isNull(e)) {
                    // 该方法在 Producer 收到 ack 时调用，为异步调用
                    // 没有异常,输出信息到控制台
                    log.info(" 主题： " + metadata.topic() + "->" + "分区：" + metadata.partition());
                } else {
                    //  出现异常打印
                    log.info("出现异常：" + e.getMessage());
                    e.printStackTrace();
                }
            });
            // 延迟一会会看到数据发往不同分区
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        // 6、关闭资源
        kafkaProducer.close();
    }
}
